package com.xrui.hbase.filter;

import com.google.common.base.Objects;

import com.xrui.hbase.ColumnName;
import com.xrui.hbase.exception.NoSuchColumnException;
import org.apache.hadoop.hbase.filter.Filter;

import java.io.IOException;

/**
 * Column filter that bounds the range of returned qualifiers within a map-type family.
 */

public class ColumnRangeFilter extends ColumnFilter {
    private static final long serialVersionUID = 1L;

    /**
     * Qualifier lower bound, or null.
     */
    private final String mMinQualifier;

    /**
     * Qualifier upper bound, or null.
     */
    private final String mMaxQualifier;

    /**
     * Whether to include the lower bound qualifier.
     */
    private final boolean mIncludeMin;

    /**
     * Whether to include the upper bound qualifier.
     */
    private final boolean mIncludeMax;

    /**
     * Initialize pagination filter with limit, offset, and other filters to fold in.
     *
     * @param min        Qualifier lower bound. Null means none.
     * @param includeMin Whether to include the lower bound.
     * @param max        Qualifier upper bound. Null means none.
     * @param includeMax Whether to include the upper bound.
     */
    public ColumnRangeFilter(String min, boolean includeMin, String max, boolean includeMax) {
        mMinQualifier = min;
        mIncludeMin = includeMin;
        mMaxQualifier = max;
        mIncludeMax = includeMax;
    }

    /**
     * Creates a column filter accepting qualifiers strictly greater than a given lower bound.
     *
     * @param qualifier Strict lower bound on accepted qualifiers.
     * @return a column filter accepting qualifiers strictly greater than a given lower bound.
     */
    public static ColumnRangeFilter greaterThan(String qualifier) {
        return new ColumnRangeFilter(qualifier, false, null, false);
    }

    /**
     * Creates a column filter accepting qualifiers greater than or equal to a given lower bound.
     *
     * @param qualifier Inclusive lower bound on accepted qualifiers.
     * @return a column filter accepting qualifiers greater than or equal to a given lower bound.
     */
    public static ColumnRangeFilter greaterThanOrEqualsTo(String qualifier) {
        return new ColumnRangeFilter(qualifier, true, null, false);
    }

    /**
     * Creates a column filter accepting qualifiers strictly smaller than a given upper bound.
     *
     * @param qualifier Strict upper bound on accepted qualifiers.
     * @return a column filter accepting qualifiers strictly smaller than a given upper bound.
     */
    public static ColumnRangeFilter lessThan(String qualifier) {
        return new ColumnRangeFilter(null, false, qualifier, false);
    }

    /**
     * Creates a column filter accepting qualifiers smaller than or equal to a given upper bound.
     *
     * @param qualifier Inclusive upper bound on accepted qualifiers.
     * @return a column filter accepting qualifiers smaller than or equal to a given upper bound.
     */
    public static ColumnRangeFilter lessThanOrEqualsTo(String qualifier) {
        return new ColumnRangeFilter(null, false, qualifier, true);
    }

    /**
     * Gets the UTF-8 encoded byte representation of a HBase column qualifier, potentially null.
     *
     * @param family    HBase column family.
     * @param qualifier HBase column qualifier, potentially null.
     * @return the HBase qualifier for the specified HBase column, or null.
     * @throws NoSuchColumnException if there is no such column.
     */
    private static byte[] toHBaseQualifierBytesOrNull(
        String family, String qualifier) throws NoSuchColumnException {
        if (qualifier == null) {
            return null;
        }
        return ColumnName.create(family, qualifier).getQualifierBytes();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Filter toHBaseFilter(ColumnName ColumnName/*, Context context*/) throws IOException {
        final String family = ColumnName.getFamily();
        return new org.apache.hadoop.hbase.filter.ColumnRangeFilter(
            toHBaseQualifierBytesOrNull(family, mMinQualifier), mIncludeMin,
            toHBaseQualifierBytesOrNull(family, mMaxQualifier), mIncludeMax);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object object) {
        if (!(object instanceof ColumnRangeFilter)) {
            return false;
        }
        final ColumnRangeFilter that = (ColumnRangeFilter) object;
        return Objects.equal(this.mMinQualifier, that.mMinQualifier)
            && Objects.equal(this.mMaxQualifier, that.mMaxQualifier)
            && Objects.equal(this.mIncludeMin, that.mIncludeMin)
            && Objects.equal(this.mIncludeMax, that.mIncludeMax);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return Objects.hashCode(mMinQualifier, mMaxQualifier, mIncludeMin, mIncludeMax);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return Objects.toStringHelper(ColumnRangeFilter.class)
            .add("min", mMinQualifier)
            .add("include-min", mIncludeMin)
            .add("max", mMaxQualifier)
            .add("include-max", mIncludeMax)
            .toString();
    }

    /**
     * Get the minimum qualifier for this column range filter.
     *
     * @return The minimum qualifier for the filter.
     */
    public String getMinQualifier() {
        return mMinQualifier;
    }

    /**
     * Get the maximum qualifier for this column range filter.
     *
     * @return The maximum qualifier for the filter.
     */
    public String getMaxQualifier() {
        return mMaxQualifier;
    }

    /**
     * Get whether to include the minimum qualifier.
     *
     * @return whether to include the minimum qualifier.
     */
    public boolean isIncludeMin() {
        return mIncludeMin;
    }

    /**
     * Get whether to include the maximum qualifier.
     *
     * @return whether to include the maximum qualifier.
     */
    public boolean isIncludeMax() {
        return mIncludeMax;
    }
}
